package com.atguigu.bigdata.spark.streaming

import org.apache.spark.SparkConf
import org.apache.spark.streaming.dstream.{DStream, ReceiverInputDStream}
import org.apache.spark.streaming.{Seconds, StreamingContext, StreamingContextState}

/**
  * create by undeRdoG on  2021-06-19  18:02
  * 凡心所向，素履以往，生如逆旅，一苇以航。
  */
object SparkStreaming07_Stop {

  def main(args: Array[String]): Unit = {
    // 创建环境
    /**
      * StreamingContext  创建时需要两个参数
      * 第一个参数环境配置
      * 第二个参数 批处理周期  duration
      **/
    val sparkConf = new SparkConf().setAppName("Streaming").setMaster("local[*]")
    val ssc = new StreamingContext(sparkConf, Seconds(3))


    // 获取端口数据
    val lines: ReceiverInputDStream[String] = ssc.socketTextStream("hadoop102", 9999)

    val wordToOne: DStream[(String, Int)] = lines.map((_, 1))

    /**
      * 窗口的范围必须是采集周期的整数倍
      * 第二个参数为滑动步长
      **/
    val value: DStream[(String, Int)] = wordToOne.window(Seconds(6), Seconds(6))

    value.reduceByKey(_ + _)

    /**
      * 由于sparkStreaming 采集器是长期执行的任务，所以不能直接关闭
      * 如果main方法执行完毕，程序也会结束
      * 因此不能让main方法执行完毕
      **/

    // 启动采集器
    ssc.start()


    /**
      * 如果想要关闭采集器，需要创建新的线程，因为  ssc.awaitTermination()  会一直 block main线程
      *
      **/

    new Thread(new Runnable {
      override def run(): Unit = {
        /**
          * 优雅的关闭：
          *
          * 关闭条件不应该由当先线程或者main线程来维护，而是应该由第三方来维护
          * 如：MySQL   Redis   zk  hdfs 等
          **/
        while (true){
          // 此处应该从第三方中拿到标志
          if (true){
            val state: StreamingContextState = ssc.getState()
            if (state == StreamingContextState.ACTIVE){
              ssc.stop(true, true)
            }
          }else{
            Thread.sleep(5000)
          }
        }
      }
    }).run()


    // 等待采集器关闭
    ssc.awaitTermination()    //  该方法会一直将 main 线程 block住


  }

}
